package pl.edu.agh.soa.monitoring.client;

import java.util.Date;
import java.util.Random;

import javax.jbi.messaging.InOnly;
import javax.jbi.messaging.MessagingException;
import javax.jbi.messaging.NormalizedMessage;
import javax.jbi.servicedesc.ServiceEndpoint;
import javax.xml.namespace.QName;

import org.apache.log4j.Logger;
import org.apache.servicemix.client.ServiceMixClient;


public class SenderThread extends Thread {
	
	private Logger logger = Logger.getLogger(this.getClass().getName());
	
	ServiceMixClient client;
	int rate;
	boolean suspended = true;
	boolean  notDestroy = true;
	Random random = new Random(new Date().getTime());
	int packageSize;
	
	public SenderThread(ServiceMixClient client, int rate,int packageSize){
		this.client = client;
		this.rate = rate;
		this.packageSize = packageSize;
	}
	
	public synchronized void kill(){
		notDestroy = false;
		suspended= false;
		this.notifyAll();
	}
	
	public void changeState(){
		synchronized(this){
			suspended = !suspended;
			this.notify();
			logger.info("State changed - suspended :" + suspended);
		}
	}
	
	public void changeRate(int newRate, int size){
		synchronized(this){
			this.rate = newRate;
			this.packageSize = size;
		}
	}
	
	public void run(){
		
			while(true){
				while (suspended){
					synchronized(this){
						try {
							this.wait();
						} catch (InterruptedException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
				}
				if (!notDestroy){
					break;
				}
				try {
					sendDataPacket();
				} catch (MessagingException e) {
					e.printStackTrace();
					break;
				}
		}
	}
	
	private String createStringContent(){
		StringBuffer sb = new StringBuffer(packageSize);
		for(int i = 0; i < packageSize; i++){
			sb.append("A");
		}
		return sb.toString();
	}
	
	private void sendDataPacket() throws MessagingException{
		InOnly exchange;
		try {
			QName service = new QName("http://pl.edu.agh.soa.esb", "receiverService");
			long before = System.currentTimeMillis();
			long distance = 0;
			logger.info("Trying to send: " + rate +" messages.");
			for (int i = 0; i < rate; i++) {
				exchange = client.createInOnlyExchange();
				exchange.setService(service);
				NormalizedMessage message = exchange.getInMessage();
//				MarketData md = new MarketData();
//				md.setTicker(createStringContent());
//				md.setPrice(Symbols.nextPrice(md.getPrice()));
//				md.setVolume(Symbols.nextVolume(10));
//				md.setTime(System.currentTimeMillis());
//				message.setProperty("data", md);
				message.setProperty("date", System.currentTimeMillis());
				long beforeInner = System.currentTimeMillis();
				client.sendSync(exchange);
				distance += System.currentTimeMillis() - beforeInner;
			}
			long after = System.currentTimeMillis();
			int sleepTime = random.nextInt(1000);
			logger.info(rate + " events sended in " + (after - before) + " ms. Sleeping : " + sleepTime + ", only sending: " + distance);
			Thread.sleep(sleepTime);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
}
